package com.flea.hclv.util;


import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.util.Map;
import java.util.Properties;


public class OAkafka implements Runnable{
    /**参数*/
    private Map<String, Object> map;
    /**参数*/
    private String topics;  
    /**日志*/
     Logger logger = LoggerFactory.getLogger(OAkafka.class);
    /**
     * 
     *@constructor 构造方法
     *@param map 参数
     **@param topics 参数
     */
    public OAkafka(Map<String, Object> map,String topics) {
        this.map =map;
        this.topics=topics;
    }
    
    /**
     * 
     *@constructor 构造方法
     */
    public OAkafka() {
    }


    public Map<String, Object> getMap() {
        return map;
    }
   

    public String getTopics() {
        return topics;
    }
    public void setTopics(String topics) {
        this.topics = topics;
    }
    public void setMap(Map<String, Object> map) {
        this.map = map;
    }
    
    
    public Properties getprops(){
        final Properties props=new Properties() ;
        try{
            final String config="producer.properties";
            final InputStream in= OAkafka.class.getClassLoader().getResourceAsStream(config);
            props.load(in);
            in.close();
        }catch(Exception e){
            e.printStackTrace();
        }
        return props;
    }     
    public void sendkafka(Map<String,Object> map,String topics){
        try {
            final KafkaProducer<String, String> producer = new KafkaProducer<String, String>(getprops());
            producer.send(new ProducerRecord<String, String>(topics, JSONObject.toJSONString(map)));
            producer.close();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }    
    }
    
    @Override
    public void run() {
        sendkafka(map,topics);
    }    
}
